package State;

import beans.SenSorReading;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 只针对key生效
 * 比如本体的计算key相同的个数
 */
public class State_KeyedState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        //定义一个有状态Map操作,统计当前sensor数据个数
        SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
                .map(new MyKeyCountMapper());

        resultStream.print();

        env.execute();
    }

    //自定义RichMapFunction
    public static class MyKeyCountMapper extends RichMapFunction<SenSorReading, Integer> {

        private ValueState<Integer> keyCountState;
        //其他类型状态的声明
//        private ListState<String> myListState;
//        private MapState<String, Double> myMapState;
//
//        private ReducingState<SenSorReading> myReducing;

        //getRuntimeContext是在open方法之后才生成的
        @Override
        public void open(Configuration parameters) throws Exception {
            //三个参数分别是名称、数据类型和初始值
            keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count", Integer.class, 0));
//            myListState = getRuntimeContext().getListState(new ListStateDescriptor<String>("my-list", String.class));
//            myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>("my-map", String.class, Double.class));
//            myReducing = getRuntimeContext().getMapState(new ReducingStateDescriptor<SenSorReading>("my-reducing", SenSorReading.class));
        }

        @Override
        public Integer map(SenSorReading senSorReading) throws Exception {
            Integer count = keyCountState.value();
            count++;
            keyCountState.update(count);

            //其他状态的api调用
//            Iterable<String> strings = myListState.get();
//            myListState.add("hello");
//
//            myMapState.put("a", 1.1);
//
//            myReducing.add(senSorReading);
            return count;
        }
    }
}
